
package com.bangcommunity.bbframe.springredis.mq;

import com.bangcommunity.bbframe.common.base.MainConfigLoader;
import com.bangcommunity.bbframe.common.utils.thread.CommonThreadPoolManager;
import com.bangcommunity.bbframe.springredis.RedisClient;
import com.bangcommunity.bbframe.springredis.mq.task.AbstractTask;
import com.bangcommunity.bbframe.springredis.mq.task.Tasks;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * 任务线程的基类.
 * 
 * @author tanghc
 */
public abstract class RedisMqConsumer<T> implements Runnable,ApplicationContextAware,InitializingBean {

    private Logger logger = LoggerFactory.getLogger(this.getClass());
    private ApplicationContext applicationContext;
    public RedisMqConsumer() {

    }

    private void init() {

        try {
            logger.info(" start init task:key={}", getRedisKey());
            CommonThreadPoolManager.getInstance().addThread(this);
            logger.info(" end init task succ:key={}", getRedisKey());
        } catch (Exception e) {
            logger.error("init task exception:key={}", getRedisKey(), e);
        }
    }

    @Override
    @SuppressWarnings("unchecked")
    public void run() {

        while (!Thread.currentThread().isInterrupted()) {
            try {
                final String key = getRedisKey();
                logger.info("~~~~~~~~~~~~redis task~~~~~~~~~~~~key:{}", key);
                ApplicationContext context = getApplicationContext();
                if (null == context) {
                    logger.error("no springcontext for task {}", this.getClass());
                    return;
                }
                // 非阻塞方式
                // T object = redisTemplate.boundListOps(key).rightPop();
                // 阻塞方式
                T object = RedisClient.lPop(key, 10, TimeUnit.SECONDS);
//                T object = RedisClient.execute(new RedisCallback<T>() {
//
//                    @Override
//                    public T doInRedis(RedisConnection conn) throws DataAccessException {
//
//                        List<byte[]> list = conn.bRPop(10000, key.getBytes());
//                        if (list == null) {
//                            logger.debug("redis task~ get nothing in redis queue: key={}", key);
//                            return null;
//                        }
//                        /**
//                         * 列表第一个为key，第二个为value
//                         */
//                        // final String key = (String)
//                        // redisTemplate.getKeySerializer().deserialize(list.get(0));
//                        final T object = (T) RedisClient.getRedisTemplate().getValueSerializer()
//                                .deserialize(list.get(1));
//                        // logger.debug("key={},object={}",key,object);
//                        return object;
//                    }
//
//                });
                if (object != null) {
                    logger.debug("redis task~ get object in redis queue: key={},object={}", key, object);
                    List<Class<? extends AbstractTask>> taskList = getTasks();
                    for (Class<? extends AbstractTask> taskClass : taskList) {
                        AbstractTask task = context.getBean(taskClass);
                        if (null != task) {
                            task.setData(object);
                            CommonThreadPoolManager.getInstance().addThread(task);
                        }
                    }
                } else {
                    logger.debug("redis task~ get object null in redis queue: key={}", key);
                }
                Thread.sleep(1);
            } catch (Throwable e) {
                logger.error("任务线程运行失败: key={}", getRedisKey(), e);
            }
        }
    }

    private String getRedisKey() {

        return new StringBuilder().append(MainConfigLoader.getInstance().getClientCode()).append(":").append(getMqKey())
                .toString();
    }

    abstract protected String getMqKey();


    @SuppressWarnings("unchecked")
    private List<Class<? extends AbstractTask>> getTasks() {

        Tasks tasks = this.getClass().getAnnotation(Tasks.class);
        Class[] classes = tasks.value();
        List<Class<? extends AbstractTask>> taskList = new ArrayList<>();
        for (Class task : classes) {
            taskList.add(task);
        }
        return taskList;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException{
        this.applicationContext = applicationContext;
    }
    public ApplicationContext getApplicationContext(){
        return this.applicationContext;
    }
    @Override
    public void afterPropertiesSet() throws Exception{
        init();
    }

}
